package com.atguigu.mr.sort;

import com.atguigu.mr.flow.FlowBean;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @description: 进行分区-在分区中又进行排序
 * @time: 20191226 18:03
 * @author: baojinlong
 **/
public class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text> {
    private FlowBean flowBean = new FlowBean();
    private Text v = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 1 获取一行
        String line = value.toString();
        // 2 截取
        String[] fields = line.split("\t");
        // 3 封装对象
        String phoneNbr = fields[0];
        long upFlow = Long.parseLong(fields[1]);
        long downFlow = Long.parseLong(fields[2]);
        assembleSumValue(flowBean, upFlow, downFlow);
        v.set(phoneNbr);
        // 4 输出
        context.write(flowBean, v);
    }

    /**
     * 封装当前value的值
     *
     * @param nowValue
     * @param sumUpFlow
     * @param sumDownFlow
     */
    private void assembleSumValue(FlowBean nowValue, long sumUpFlow, long sumDownFlow) {
        nowValue.setUpFlow(sumUpFlow);
        nowValue.setDownFlow(sumDownFlow);
        nowValue.setSumFlow(sumUpFlow + sumDownFlow);
    }
}
